package com.lagou.spark
/*
2、日志分析
任务：
2.1、计算独立IP数
2.2、统计每个视频独立IP数（视频的标志：在日志文件的某些可以找到 *.mp4，代表一个视频文件）
2.3、统计一天中每个小时的流量
 */
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object LogAnalysis {
  def main(args: Array[String]): Unit = {
    //创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    //读取文件
    val line: RDD[String] = sc.textFile("data/cdn.txt")
    val value: RDD[Array[String]] = line.map(_.split("\\s+"))

    //1、计算独立ip数
    val independentIP: Long = value.map(x => x(0)).distinct().count()
    println("独立IP数：" + independentIP)
    println("-----------------------------------------------------")

    //2、统计每个视频独立IP数
    //从cdn.txt -> get url中使用正则表达式获取.mp4
    val video: collection.Map[Option[String], Long] = value.map(x => ("\\d+\\.mp4".r.findFirstIn(x(6)), x(0)))
      .filter(_._1 != " ") 
      .distinct()
      .countByKey() //按.mp4数据进行统计
    video.foreach(println)
    println("-----------------------------------------------------")

    //3、统计一天中每个小时的流量
    val times: RDD[(String, Double)] = line.map(x => {
      val t: Array[String] = x.split("\\s+")
      //以切分后的时间字段，尾追加继续切分，得到年、月、日、时
      val year: String = t(3).tail.split("[/:]")(2)
      val month: String = t(3).tail.split("[/:]")(1)
      val day: String = t(3).tail.split("[/:]")(0)
      val hour: String = t(3).tail.split("[/:]")(3)
      val time: String = year + "/" + month + "/" + day + "/" + hour + ":00"
      (time, t(9).toDouble)
    })
      //仅过滤出2017/Feb/15这一天的数据做聚合
      .filter {
       case (k, v) => k.contains("2017/Feb/15")
      }
      .reduceByKey(_ + _)
    times.collect().foreach(println)

    //关闭SparkContext
    sc.stop()
  }
}
